package com.ljm.boot.rabbitmq.producer;

import com.ljm.boot.rabbitmq.constant.*;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

@Component
public class DirectProducer {

    @Resource
    private RabbitTemplate rabbitTemplate;


    @PostConstruct
    public void sendMsg(){
        String msg = "first msg";
        System.out.println("发送消息-" + msg);
        rabbitTemplate.convertAndSend("amq.direct","directok",msg);
    }



    public void sendDirect1(Object msg){
        System.out.println("发送消息-" + msg);
        rabbitTemplate.convertAndSend(RabbitMQConstant.DIRECT_EXCHANGE,"queue.red",msg);
    }

    public void sendDirect2(Object msg){
        System.out.println("发送消息-" + msg);
        rabbitTemplate.convertAndSend(RabbitMQConstant.DIRECT_EXCHANGE,"queue.green",msg);
    }

    //保证数据的可靠性
    public void test(){
        //创建一个CorrelationData Correlation：关联
        CorrelationData correlationData = new CorrelationData();
        //设置消息的id
        correlationData.setId("msg-888888");
        //准备好 回调函数:Callback，用于接收 将来:Future 的确认结果
        correlationData.getFuture().addCallback(
                /**
                 * 当消息发送没有出现异常时，这个方法会被调用
                 */
                result -> {
                    if (result.isAck()){
                        System.out.println("发送消息成功，消息已到达交换机，消息id:"+correlationData.getId());
                    }else {
                        System.out.println("发送消息成功，消息没有到达交换机，消息id:"+correlationData.getId()+" 原因："+result.getReason());

                    }                },
                /**
                 * 当消息发送出现异常时，这个方法会被调用
                 */
                ex -> {
                    System.out.println("发送消息异常，消息id:"+correlationData.getId()+" 异常："+ex);
                }
        );
        //自己随便定一个没有的交换机
        rabbitTemplate.convertAndSend(RabbitMQConstant.DIRECT_EXCHANGE, "queue.green", "到达交换机了吗",correlationData);
        //rabbitTemplate.setConfirmCallback(new DirectCallback());
        rabbitTemplate.setReturnsCallback(new DirectCallback());
    }

    //延时队列
    public void testDelayQueue() {
        String message = "hello,delayed message";
        //2.发送消息，利用消息后置处理器添加消息头
        rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //添加延迟消息属性
                message.getMessageProperties().setDelay(5000);
                return message;
            }
        });

    }
}
